/**
 * Copyright The Apache Software Foundation
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase;

import static org.apache.hadoop.hbase.util.Bytes.len;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.util.ByteBufferUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import org.apache.hadoop.io.RawComparator;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;

/**
 * An HBase Key/Value. This is the fundamental HBase Type.
 *
 * HBase应用程序和用户应使用Cell接口，并避免直接使用Cell中未定义的KeyValue和成员函数。
 * HBase applications and users should use the Cell interface and avoid directly using KeyValue and
 * member functions not defined in Cell.
 *
 * 如果在客户端使用，则访问各个字段的主要方法是 #getRowArray()，#getFamilyArray()，#getQualifierArray()，
 * #getTimestamp() 和 #getValueArray()。
 * 这些方法分配新的字节数组并返回副本。避免在服务器端使用它们。
 * If being used client-side, the primary methods to access individual fields are
 * {@link #getRowArray()}, {@link #getFamilyArray()}, {@link #getQualifierArray()},
 * {@link #getTimestamp()}, and {@link #getValueArray()}. These methods allocate new byte arrays
 * and return copies. Avoid their use server-side.
 *
 * 此类的实例是不可变的。它们没有实现Comparable，但是提供了Comparators。
 * 比较器随上下文而变化，无论是用户表比较还是目录表比较。
 * 关键是您使用适当的比较器。有用于普通HFiles，Meta的 Hfiles和 Bloom筛选键的比较器。
 * Instances of this class are immutable. They do not implement Comparable but Comparators are provided.
 * Comparators change with context, whether user table or a catalog table comparison.
 * Its critical you use the appropriate comparator. There are Comparators for normal HFiles, Meta's Hfiles, and bloom filter keys.
 *
 *
 * KeyValue wraps a byte array and takes offsets and lengths into passed array at where to start
 * interpreting the content as KeyValue. The KeyValue format inside a byte array is:
 * <code>
 * &lt;keylength&gt;
 * &lt;valuelength&gt;
 * &lt;key&gt;
 * &lt;value&gt;
 * </code>
 *
 * Key is further decomposed as:
 * <code>
 * &lt;rowlength&gt;
 * &lt;row&gt;
 * &lt;columnfamilylength&gt;
 * &lt;columnfamily&gt;
 * &lt;columnqualifier&gt;
 * &lt;timestamp&gt;
 * &lt;keytype&gt;
 * </code>
 *
 * TODO_MA 总结：
 * HBase内部，单元格Cell的实现为KeyValue，它是HBase某行数据的某个单元格在内存中的组织形式，由Key Length、Value Length、Key、Value四大部分组成。
 * 其中，Key又由Row Length、Row、Column Family Length、Column Family、Column Qualifier、Time Stamp、Key Type七部分组成。
 *
 * The <code>rowlength</code> maximum is <code>Short.MAX_SIZE</code>,
 * column family length maximum is <code>Byte.MAX_SIZE</code>,
 * and column qualifier + key length must be &lt; <code>Integer.MAX_SIZE</code>.
 *
 * The column does not contain the family/qualifier delimiter, {@link #COLUMN_FAMILY_DELIMITER}<br>
 *
 * KeyValue can optionally contain Tags. When it contains tags, it is added in the byte array after
 * the value part. The format for this part is: <code>&lt;tagslength&gt;&lt;tagsbytes&gt;</code>.
 * <code>tagslength</code> maximum is <code>Short.MAX_SIZE</code>. The <code>tagsbytes</code>
 * contain one or more tags where as each tag is of the form
 * <code>&lt;taglength&gt;&lt;tagtype&gt;&lt;tagbytes&gt;</code>. <code>tagtype</code> is one byte
 * and <code>taglength</code> maximum is <code>Short.MAX_SIZE</code> and it includes 1 byte type
 * length and actual tag bytes length.
 *
 * CloneUtil
 */
@InterfaceAudience.Private
public class KeyValue implements ExtendedCell, Cloneable {
    private static final ArrayList<Tag> EMPTY_ARRAY_LIST = new ArrayList<>();

    private static final Logger LOG = LoggerFactory.getLogger(KeyValue.class);

    public static final int FIXED_OVERHEAD = ClassSize.OBJECT + // the KeyValue object itself
            ClassSize.REFERENCE + // pointer to "bytes"
            2 * Bytes.SIZEOF_INT + // offset, length
            Bytes.SIZEOF_LONG;// memstoreTS

    /**
     * Colon character in UTF-8
     */
    public static final char COLUMN_FAMILY_DELIMITER = ':';

    public static final byte[] COLUMN_FAMILY_DELIM_ARRAY = new byte[]{COLUMN_FAMILY_DELIMITER};

    /**
     * Comparator for plain key/values; i.e. non-catalog table key/values. Works on Key portion
     * of KeyValue only.
     *
     * @deprecated Use {@link CellComparator#getInstance()} instead. Deprecated for hbase 2.0, remove for hbase 3.0.
     */
    @Deprecated
    public static final KVComparator COMPARATOR = new KVComparator();
    /**
     * A {@link KVComparator} for <code>hbase:meta</code> catalog table
     * {@link KeyValue}s.
     *
     * @deprecated Use {@link CellComparatorImpl#META_COMPARATOR} instead. Deprecated for hbase 2.0, remove for hbase 3.0.
     */
    @Deprecated
    public static final KVComparator META_COMPARATOR = new MetaComparator();

    /**
     * Size of the key length field in bytes
     */
    // TODO_MA 注释：key：4 个字节
    public static final int KEY_LENGTH_SIZE = Bytes.SIZEOF_INT;

    /**
     * Size of the key type field in bytes
     */
    // TODO_MA 注释： 类型type： 1 个字节
    public static final int TYPE_SIZE = Bytes.SIZEOF_BYTE;

    /**
     * Size of the row length field in bytes
     */
    // TODO_MA 注释： row length = 2 个字节
    public static final int ROW_LENGTH_SIZE = Bytes.SIZEOF_SHORT;

    /**
     * Size of the family length field in bytes
     */
    // TODO_MA 注释：列簇长度： 1 个字节
    public static final int FAMILY_LENGTH_SIZE = Bytes.SIZEOF_BYTE;

    /**
     * Size of the timestamp field in bytes
     */
    // TODO_MA 注释： 时间戳：8 个字节
    public static final int TIMESTAMP_SIZE = Bytes.SIZEOF_LONG;

    // Size of the timestamp and type byte on end of a key -- a long + a byte.
    public static final int TIMESTAMP_TYPE_SIZE = TIMESTAMP_SIZE + TYPE_SIZE;

    // Size of the length shorts and bytes in key.
    public static final int KEY_INFRASTRUCTURE_SIZE = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + TIMESTAMP_TYPE_SIZE;

    // How far into the key the row starts at. First thing to read is the short that says how long the row is.
    public static final int ROW_OFFSET = Bytes.SIZEOF_INT /*keylength*/ + Bytes.SIZEOF_INT /*valuelength*/;

    public static final int ROW_KEY_OFFSET = ROW_OFFSET + ROW_LENGTH_SIZE;

    // Size of the length ints in a KeyValue datastructure.
    public static final int KEYVALUE_INFRASTRUCTURE_SIZE = ROW_OFFSET;

    /**
     * Size of the tags length field in bytes
     */
    public static final int TAGS_LENGTH_SIZE = Bytes.SIZEOF_SHORT;

    public static final int KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE = ROW_OFFSET + TAGS_LENGTH_SIZE;

    /**
     * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
     * characteristics would take up for its underlying data structure.
     *
     * @param rlength row length
     * @param flength family length
     * @param qlength qualifier length
     * @param vlength value length
     * @return the <code>KeyValue</code> data structure length
     */
    public static long getKeyValueDataStructureSize(int rlength, int flength, int qlength, int vlength) {
        return KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + getKeyDataStructureSize(rlength, flength, qlength) + vlength;
    }

    /**
     * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
     * characteristics would take up for its underlying data structure.
     *
     * @param rlength    row length
     * @param flength    family length
     * @param qlength    qualifier length
     * @param vlength    value length
     * @param tagsLength total length of the tags
     * @return the <code>KeyValue</code> data structure length
     */
    public static long getKeyValueDataStructureSize(int rlength, int flength, int qlength, int vlength, int tagsLength) {
        if(tagsLength == 0) {
            return getKeyValueDataStructureSize(rlength, flength, qlength, vlength);
        }
        return KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE + getKeyDataStructureSize(rlength, flength, qlength) + vlength + tagsLength;
    }

    /**
     * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
     * characteristics would take up for its underlying data structure.
     *
     * @param klength    key length
     * @param vlength    value length
     * @param tagsLength total length of the tags
     * @return the <code>KeyValue</code> data structure length
     */
    public static long getKeyValueDataStructureSize(int klength, int vlength, int tagsLength) {
        if(tagsLength == 0) {
            return (long) KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE + klength + vlength;
        }
        return (long) KeyValue.KEYVALUE_WITH_TAGS_INFRASTRUCTURE_SIZE + klength + vlength + tagsLength;
    }

    /**
     * Computes the number of bytes that a <code>KeyValue</code> instance with the provided
     * characteristics would take up in its underlying data structure for the key.
     *
     * @param rlength row length
     * @param flength family length
     * @param qlength qualifier length
     * @return the key data structure length
     */
    public static long getKeyDataStructureSize(int rlength, int flength, int qlength) {
        return (long) KeyValue.KEY_INFRASTRUCTURE_SIZE + rlength + flength + qlength;
    }

    /**
     * Key type.
     * Has space for other key types to be added later.  Cannot rely on
     * enum ordinals . They change if item is removed or moved.  Do our own codes.
     */
    public static enum Type {
        Minimum((byte) 0), Put((byte) 4),

        Delete((byte) 8), DeleteFamilyVersion((byte) 10), DeleteColumn((byte) 12), DeleteFamily((byte) 14),

        // Maximum is used when searching; you look from maximum on down.
        Maximum((byte) 255);

        private final byte code;

        Type(final byte c) {
            this.code = c;
        }

        public byte getCode() {
            return this.code;
        }

        private static Type[] codeArray = new Type[256];

        static {
            for(Type t : Type.values()) {
                codeArray[t.code & 0xff] = t;
            }
        }

        /**
         * True to indicate that the byte b is a valid type.
         *
         * @param b byte to check
         * @return true or false
         */
        static boolean isValidType(byte b) {
            return codeArray[b & 0xff] != null;
        }

        /**
         * Cannot rely on enum ordinals . They change if item is removed or moved.
         * Do our own codes.
         *
         * @param b
         * @return Type associated with passed code.
         */
        public static Type codeToType(final byte b) {
            Type t = codeArray[b & 0xff];
            if(t != null) {
                return t;
            }
            throw new RuntimeException("Unknown code " + b);
        }
    }

    /**
     * Lowest possible key.
     * Makes a Key with highest possible Timestamp, empty row and column.  No
     * key can be equal or lower than this one in memstore or in store file.
     */
    public static final KeyValue LOWESTKEY = new KeyValue(HConstants.EMPTY_BYTE_ARRAY, HConstants.LATEST_TIMESTAMP);


    /********
     * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
     *   注释：
     *   1、包含真是 key-value 的一个不可变字节数组
     *   2、当前这个 key-value 在 bytesbuffer 中的起始位移
     *   3、当前这个 key-value 的长度
     */
    // KeyValue core instance fields.
    protected byte[] bytes = null;          // an immutable byte array that contains the KV
    protected int offset = 0;               // offset into bytes buffer KV starts at
    protected int length = 0;               // length of the KV starting from offset.

    /** Here be dragons **/

    /**
     * used to achieve atomic operations in the memstore.
     */
    @Override
    public long getSequenceId() {
        return seqId;
    }

    @Override
    public void setSequenceId(long seqId) {
        this.seqId = seqId;
    }

    // multi-version concurrency control version.  default value is 0, aka do not care.
    private long seqId = 0;

    /** Dragon time over, return to normal business */


    /**
     * Writable Constructor -- DO NOT USE
     */
    public KeyValue() {
    }

    /**
     * Creates a KeyValue from the start of the specified byte array.
     * Presumes <code>bytes</code> content is formatted as a KeyValue blob.
     *
     * @param bytes byte array
     */
    public KeyValue(final byte[] bytes) {
        this(bytes, 0);
    }

    /**
     * Creates a KeyValue from the specified byte array and offset.
     * Presumes <code>bytes</code> content starting at <code>offset</code> is
     * formatted as a KeyValue blob.
     *
     * @param bytes  byte array
     * @param offset offset to start of KeyValue
     */
    public KeyValue(final byte[] bytes, final int offset) {
        this(bytes, offset, getLength(bytes, offset));
    }

    /**
     * Creates a KeyValue from the specified byte array, starting at offset, and
     * for length <code>length</code>.
     *
     * @param bytes  byte array
     * @param offset offset to start of the KeyValue
     * @param length length of the KeyValue
     */
    public KeyValue(final byte[] bytes, final int offset, final int length) {
        KeyValueUtil.checkKeyValueBytes(bytes, offset, length, true);
        this.bytes = bytes;
        this.offset = offset;
        this.length = length;
    }

    /**
     * Creates a KeyValue from the specified byte array, starting at offset, and
     * for length <code>length</code>.
     *
     * @param bytes  byte array
     * @param offset offset to start of the KeyValue
     * @param length length of the KeyValue
     * @param ts
     */
    public KeyValue(final byte[] bytes, final int offset, final int length, long ts) {
        this(bytes, offset, length, null, 0, 0, null, 0, 0, ts, Type.Maximum, null, 0, 0, null);
    }

    /** Constructors that build a new backing byte array from fields */

    /**
     * Constructs KeyValue structure filled with null value.
     * Sets type to {@link KeyValue.Type#Maximum}
     *
     * @param row       - row key (arbitrary byte array)
     * @param timestamp
     */
    public KeyValue(final byte[] row, final long timestamp) {
        this(row, null, null, timestamp, Type.Maximum, null);
    }

    /**
     * Constructs KeyValue structure filled with null value.
     *
     * @param row       - row key (arbitrary byte array)
     * @param timestamp
     */
    public KeyValue(final byte[] row, final long timestamp, Type type) {
        this(row, null, null, timestamp, type, null);
    }

    /**
     * Constructs KeyValue structure filled with null value.
     * Sets type to {@link KeyValue.Type#Maximum}
     *
     * @param row       - row key (arbitrary byte array)
     * @param family    family name
     * @param qualifier column qualifier
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier) {
        this(row, family, qualifier, HConstants.LATEST_TIMESTAMP, Type.Maximum);
    }

    /**
     * Constructs KeyValue structure as a put filled with specified values and
     * LATEST_TIMESTAMP.
     *
     * @param row       - row key (arbitrary byte array)
     * @param family    family name
     * @param qualifier column qualifier
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final byte[] value) {
        this(row, family, qualifier, HConstants.LATEST_TIMESTAMP, Type.Put, value);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param type      key type
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, Type type) {
        this(row, family, qualifier, timestamp, type, null);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param value     column value
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, final byte[] value) {
        this(row, family, qualifier, timestamp, Type.Put, value);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param value     column value
     * @param tags      tags
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, final byte[] value, final Tag[] tags) {
        this(row, family, qualifier, timestamp, value, tags != null ? Arrays.asList(tags) : null);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param value     column value
     * @param tags      tags non-empty list of tags or null
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, final byte[] value, final List<Tag> tags) {
        this(row, 0, row == null ? 0 : row.length, family, 0, family == null ? 0 : family.length, qualifier, 0,
                qualifier == null ? 0 : qualifier.length, timestamp, Type.Put, value, 0, value == null ? 0 : value.length, tags);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, Type type, final byte[] value) {
        this(row, 0, len(row), family, 0, len(family), qualifier, 0, len(qualifier), timestamp, type, value, 0, len(value));
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, Type type, final byte[] value,
            final List<Tag> tags) {
        this(row, family, qualifier, 0, qualifier == null ? 0 : qualifier.length, timestamp, type, value, 0, value == null ? 0 : value.length, tags);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final byte[] family, final byte[] qualifier, final long timestamp, Type type, final byte[] value,
            final byte[] tags) {
        this(row, family, qualifier, 0, qualifier == null ? 0 : qualifier.length, timestamp, type, value, 0, value == null ? 0 : value.length, tags);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     *
     * @param row       row key
     * @param family    family name
     * @param qualifier column qualifier
     * @param qoffset   qualifier offset
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @param voffset   value offset
     * @param vlength   value length
     * @throws IllegalArgumentException
     */
    public KeyValue(byte[] row, byte[] family, byte[] qualifier, int qoffset, int qlength, long timestamp, Type type, byte[] value, int voffset,
            int vlength, List<Tag> tags) {
        this(row, 0, row == null ? 0 : row.length, family, 0, family == null ? 0 : family.length, qualifier, qoffset, qlength, timestamp, type, value,
                voffset, vlength, tags);
    }

    /**
     * @param row
     * @param family
     * @param qualifier
     * @param qoffset
     * @param qlength
     * @param timestamp
     * @param type
     * @param value
     * @param voffset
     * @param vlength
     * @param tags
     */
    public KeyValue(byte[] row, byte[] family, byte[] qualifier, int qoffset, int qlength, long timestamp, Type type, byte[] value, int voffset,
            int vlength, byte[] tags) {
        this(row, 0, row == null ? 0 : row.length, family, 0, family == null ? 0 : family.length, qualifier, qoffset, qlength, timestamp, type, value,
                voffset, vlength, tags, 0, tags == null ? 0 : tags.length);
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param row row key
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset, final int flength,
            final byte[] qualifier, final int qoffset, final int qlength, final long timestamp, final Type type, final byte[] value,
            final int voffset, final int vlength) {
        this(row, roffset, rlength, family, foffset, flength, qualifier, qoffset, qlength, timestamp, type, value, voffset, vlength, null);
    }

    /**
     * Constructs KeyValue structure filled with specified values. Uses the provided buffer as the
     * data buffer.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param buffer    the bytes buffer to use
     * @param boffset   buffer offset
     * @param row       row key
     * @param roffset   row offset
     * @param rlength   row length
     * @param family    family name
     * @param foffset   family offset
     * @param flength   family length
     * @param qualifier column qualifier
     * @param qoffset   qualifier offset
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @param voffset   value offset
     * @param vlength   value length
     * @param tags      non-empty list of tags or null
     * @throws IllegalArgumentException an illegal value was passed or there is insufficient space
     *                                  remaining in the buffer
     */
    public KeyValue(byte[] buffer, final int boffset, final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset,
            final int flength, final byte[] qualifier, final int qoffset, final int qlength, final long timestamp, final Type type,
            final byte[] value, final int voffset, final int vlength, final Tag[] tags) {
        this.bytes = buffer;
        this.length = writeByteArray(buffer, boffset, row, roffset, rlength, family, foffset, flength, qualifier, qoffset, qlength, timestamp, type,
                value, voffset, vlength, tags);
        this.offset = boffset;
    }

    /**
     * Constructs KeyValue structure filled with specified values.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param row       row key
     * @param roffset   row offset
     * @param rlength   row length
     * @param family    family name
     * @param foffset   family offset
     * @param flength   family length
     * @param qualifier column qualifier
     * @param qoffset   qualifier offset
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @param voffset   value offset
     * @param vlength   value length
     * @param tags      tags
     * @throws IllegalArgumentException
     */
    public KeyValue(final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset, final int flength,
            final byte[] qualifier, final int qoffset, final int qlength, final long timestamp, final Type type, final byte[] value,
            final int voffset, final int vlength, final List<Tag> tags) {
        this.bytes = createByteArray(row, roffset, rlength, family, foffset, flength, qualifier, qoffset, qlength, timestamp, type, value, voffset,
                vlength, tags);
        this.length = bytes.length;
        this.offset = 0;
    }

    /**
     * @param row
     * @param roffset
     * @param rlength
     * @param family
     * @param foffset
     * @param flength
     * @param qualifier
     * @param qoffset
     * @param qlength
     * @param timestamp
     * @param type
     * @param value
     * @param voffset
     * @param vlength
     * @param tags
     */
    public KeyValue(final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset, final int flength,
            final byte[] qualifier, final int qoffset, final int qlength, final long timestamp, final Type type, final byte[] value,
            final int voffset, final int vlength, final byte[] tags, final int tagsOffset, final int tagsLength) {
        this.bytes = createByteArray(row, roffset, rlength, family, foffset, flength, qualifier, qoffset, qlength, timestamp, type, value, voffset,
                vlength, tags, tagsOffset, tagsLength);
        this.length = bytes.length;
        this.offset = 0;
    }

    /**
     * Constructs an empty KeyValue structure, with specified sizes.
     * This can be used to partially fill up KeyValues.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param rlength   row length
     * @param flength   family length
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param vlength   value length
     * @throws IllegalArgumentException
     */
    public KeyValue(final int rlength, final int flength, final int qlength, final long timestamp, final Type type, final int vlength) {
        this(rlength, flength, qlength, timestamp, type, vlength, 0);
    }

    /**
     * Constructs an empty KeyValue structure, with specified sizes.
     * This can be used to partially fill up KeyValues.
     * <p>
     * Column is split into two fields, family and qualifier.
     *
     * @param rlength    row length
     * @param flength    family length
     * @param qlength    qualifier length
     * @param timestamp  version timestamp
     * @param type       key type
     * @param vlength    value length
     * @param tagsLength
     * @throws IllegalArgumentException
     */
    public KeyValue(final int rlength, final int flength, final int qlength, final long timestamp, final Type type, final int vlength,
            final int tagsLength) {
        this.bytes = createEmptyByteArray(rlength, flength, qlength, timestamp, type, vlength, tagsLength);
        this.length = bytes.length;
        this.offset = 0;
    }


    public KeyValue(byte[] row, int roffset, int rlength, byte[] family, int foffset, int flength, ByteBuffer qualifier, long ts, Type type,
            ByteBuffer value, List<Tag> tags) {
        this.bytes = createByteArray(row, roffset, rlength, family, foffset, flength, qualifier, 0, qualifier == null ? 0 : qualifier.remaining(), ts,
                type, value, 0, value == null ? 0 : value.remaining(), tags);
        this.length = bytes.length;
        this.offset = 0;
    }

    public KeyValue(Cell c) {
        this(c.getRowArray(), c.getRowOffset(), c.getRowLength(), c.getFamilyArray(), c.getFamilyOffset(), c.getFamilyLength(), c.getQualifierArray(),
                c.getQualifierOffset(), c.getQualifierLength(), c.getTimestamp(), Type.codeToType(c.getTypeByte()), c.getValueArray(),
                c.getValueOffset(), c.getValueLength(), c.getTagsArray(), c.getTagsOffset(), c.getTagsLength());
        this.seqId = c.getSequenceId();
    }

    /**
     * Create an empty byte[] representing a KeyValue
     * All lengths are preset and can be filled in later.
     *
     * @param rlength
     * @param flength
     * @param qlength
     * @param timestamp
     * @param type
     * @param vlength
     * @return The newly created byte array.
     */
    private static byte[] createEmptyByteArray(final int rlength, int flength, int qlength, final long timestamp, final Type type, int vlength,
            int tagsLength) {
        if(rlength > Short.MAX_VALUE) {
            throw new IllegalArgumentException("Row > " + Short.MAX_VALUE);
        }
        if(flength > Byte.MAX_VALUE) {
            throw new IllegalArgumentException("Family > " + Byte.MAX_VALUE);
        }
        // Qualifier length
        if(qlength > Integer.MAX_VALUE - rlength - flength) {
            throw new IllegalArgumentException("Qualifier > " + Integer.MAX_VALUE);
        }
        RawCell.checkForTagsLength(tagsLength);
        // Key length
        long longkeylength = getKeyDataStructureSize(rlength, flength, qlength);
        if(longkeylength > Integer.MAX_VALUE) {
            throw new IllegalArgumentException("keylength " + longkeylength + " > " + Integer.MAX_VALUE);
        }
        int keylength = (int) longkeylength;
        // Value length
        if(vlength > HConstants.MAXIMUM_VALUE_LENGTH) { // FindBugs INT_VACUOUS_COMPARISON
            throw new IllegalArgumentException("Valuer > " + HConstants.MAXIMUM_VALUE_LENGTH);
        }

        // Allocate right-sized byte array.
        byte[] bytes = new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength, tagsLength)];
        // Write the correct size markers
        int pos = 0;
        pos = Bytes.putInt(bytes, pos, keylength);
        pos = Bytes.putInt(bytes, pos, vlength);
        pos = Bytes.putShort(bytes, pos, (short) (rlength & 0x0000ffff));
        pos += rlength;
        pos = Bytes.putByte(bytes, pos, (byte) (flength & 0x0000ff));
        pos += flength + qlength;
        pos = Bytes.putLong(bytes, pos, timestamp);
        pos = Bytes.putByte(bytes, pos, type.getCode());
        pos += vlength;
        if(tagsLength > 0) {
            pos = Bytes.putAsShort(bytes, pos, tagsLength);
        }
        return bytes;
    }

    /**
     * Checks the parameters passed to a constructor.
     *
     * @param row     row key
     * @param rlength row length
     * @param family  family name
     * @param flength family length
     * @param qlength qualifier length
     * @param vlength value length
     * @throws IllegalArgumentException an illegal value was passed
     */
    static void checkParameters(final byte[] row, final int rlength, final byte[] family, int flength, int qlength,
            int vlength) throws IllegalArgumentException {
        if(rlength > Short.MAX_VALUE) {
            throw new IllegalArgumentException("Row > " + Short.MAX_VALUE);
        }
        if(row == null) {
            throw new IllegalArgumentException("Row is null");
        }
        // Family length
        flength = family == null ? 0 : flength;
        if(flength > Byte.MAX_VALUE) {
            throw new IllegalArgumentException("Family > " + Byte.MAX_VALUE);
        }
        // Qualifier length
        if(qlength > Integer.MAX_VALUE - rlength - flength) {
            throw new IllegalArgumentException("Qualifier > " + Integer.MAX_VALUE);
        }
        // Key length
        long longKeyLength = getKeyDataStructureSize(rlength, flength, qlength);
        if(longKeyLength > Integer.MAX_VALUE) {
            throw new IllegalArgumentException("keylength " + longKeyLength + " > " + Integer.MAX_VALUE);
        }
        // Value length
        if(vlength > HConstants.MAXIMUM_VALUE_LENGTH) { // FindBugs INT_VACUOUS_COMPARISON
            throw new IllegalArgumentException("Value length " + vlength + " > " + HConstants.MAXIMUM_VALUE_LENGTH);
        }
    }

    /**
     * Write KeyValue format into the provided byte array.
     *
     * @param buffer    the bytes buffer to use
     * @param boffset   buffer offset
     * @param row       row key
     * @param roffset   row offset
     * @param rlength   row length
     * @param family    family name
     * @param foffset   family offset
     * @param flength   family length
     * @param qualifier column qualifier
     * @param qoffset   qualifier offset
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @param voffset   value offset
     * @param vlength   value length
     * @return The number of useful bytes in the buffer.
     * @throws IllegalArgumentException an illegal value was passed or there is insufficient space
     *                                  remaining in the buffer
     */
    public static int writeByteArray(byte[] buffer, final int boffset, final byte[] row, final int roffset, final int rlength, final byte[] family,
            final int foffset, int flength, final byte[] qualifier, final int qoffset, int qlength, final long timestamp, final Type type,
            final byte[] value, final int voffset, int vlength, Tag[] tags) {

        checkParameters(row, rlength, family, flength, qlength, vlength);

        // Calculate length of tags area
        int tagsLength = 0;
        if(tags != null && tags.length > 0) {
            for(Tag t : tags) {
                tagsLength += t.getValueLength() + Tag.INFRASTRUCTURE_SIZE;
            }
        }
        RawCell.checkForTagsLength(tagsLength);
        int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
        int keyValueLength = (int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength, tagsLength);
        if(keyValueLength > buffer.length - boffset) {
            throw new IllegalArgumentException("Buffer size " + (buffer.length - boffset) + " < " + keyValueLength);
        }

        // Write key, value and key row length.
        int pos = boffset;
        pos = Bytes.putInt(buffer, pos, keyLength);
        pos = Bytes.putInt(buffer, pos, vlength);
        pos = Bytes.putShort(buffer, pos, (short) (rlength & 0x0000ffff));
        pos = Bytes.putBytes(buffer, pos, row, roffset, rlength);
        pos = Bytes.putByte(buffer, pos, (byte) (flength & 0x0000ff));
        if(flength != 0) {
            pos = Bytes.putBytes(buffer, pos, family, foffset, flength);
        }
        if(qlength != 0) {
            pos = Bytes.putBytes(buffer, pos, qualifier, qoffset, qlength);
        }
        pos = Bytes.putLong(buffer, pos, timestamp);
        pos = Bytes.putByte(buffer, pos, type.getCode());
        if(value != null && value.length > 0) {
            pos = Bytes.putBytes(buffer, pos, value, voffset, vlength);
        }
        // Write the number of tags. If it is 0 then it means there are no tags.
        if(tagsLength > 0) {
            pos = Bytes.putAsShort(buffer, pos, tagsLength);
            for(Tag t : tags) {
                int tlen = t.getValueLength();
                pos = Bytes.putAsShort(buffer, pos, tlen + Tag.TYPE_LENGTH_SIZE);
                pos = Bytes.putByte(buffer, pos, t.getType());
                Tag.copyValueTo(t, buffer, pos);
                pos += tlen;
            }
        }
        return keyValueLength;
    }

    /**
     * Write KeyValue format into a byte array.
     *
     * @param row       row key
     * @param roffset   row offset
     * @param rlength   row length
     * @param family    family name
     * @param foffset   family offset
     * @param flength   family length
     * @param qualifier column qualifier
     * @param qoffset   qualifier offset
     * @param qlength   qualifier length
     * @param timestamp version timestamp
     * @param type      key type
     * @param value     column value
     * @param voffset   value offset
     * @param vlength   value length
     * @return The newly created byte array.
     */
    private static byte[] createByteArray(final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset, int flength,
            final byte[] qualifier, final int qoffset, int qlength, final long timestamp, final Type type, final byte[] value, final int voffset,
            int vlength, byte[] tags, int tagsOffset, int tagsLength) {

        checkParameters(row, rlength, family, flength, qlength, vlength);
        RawCell.checkForTagsLength(tagsLength);
        // Allocate right-sized byte array.
        int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
        byte[] bytes = new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength, tagsLength)];
        // Write key, value and key row length.
        int pos = 0;
        pos = Bytes.putInt(bytes, pos, keyLength);
        pos = Bytes.putInt(bytes, pos, vlength);
        pos = Bytes.putShort(bytes, pos, (short) (rlength & 0x0000ffff));
        pos = Bytes.putBytes(bytes, pos, row, roffset, rlength);
        pos = Bytes.putByte(bytes, pos, (byte) (flength & 0x0000ff));
        if(flength != 0) {
            pos = Bytes.putBytes(bytes, pos, family, foffset, flength);
        }
        if(qlength != 0) {
            pos = Bytes.putBytes(bytes, pos, qualifier, qoffset, qlength);
        }
        pos = Bytes.putLong(bytes, pos, timestamp);
        pos = Bytes.putByte(bytes, pos, type.getCode());
        if(value != null && value.length > 0) {
            pos = Bytes.putBytes(bytes, pos, value, voffset, vlength);
        }
        // Add the tags after the value part
        if(tagsLength > 0) {
            pos = Bytes.putAsShort(bytes, pos, tagsLength);
            pos = Bytes.putBytes(bytes, pos, tags, tagsOffset, tagsLength);
        }
        return bytes;
    }

    /**
     * @param qualifier can be a ByteBuffer or a byte[], or null.
     * @param value     can be a ByteBuffer or a byte[], or null.
     */
    private static byte[] createByteArray(final byte[] row, final int roffset, final int rlength, final byte[] family, final int foffset, int flength,
            final Object qualifier, final int qoffset, int qlength, final long timestamp, final Type type, final Object value, final int voffset,
            int vlength, List<Tag> tags) {

        checkParameters(row, rlength, family, flength, qlength, vlength);

        // Calculate length of tags area
        int tagsLength = 0;
        if(tags != null && !tags.isEmpty()) {
            for(Tag t : tags) {
                tagsLength += t.getValueLength() + Tag.INFRASTRUCTURE_SIZE;
            }
        }
        RawCell.checkForTagsLength(tagsLength);
        // Allocate right-sized byte array.
        int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);
        byte[] bytes = new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength, tagsLength)];

        // Write key, value and key row length.
        int pos = 0;
        pos = Bytes.putInt(bytes, pos, keyLength);

        pos = Bytes.putInt(bytes, pos, vlength);
        pos = Bytes.putShort(bytes, pos, (short) (rlength & 0x0000ffff));
        pos = Bytes.putBytes(bytes, pos, row, roffset, rlength);
        pos = Bytes.putByte(bytes, pos, (byte) (flength & 0x0000ff));
        if(flength != 0) {
            pos = Bytes.putBytes(bytes, pos, family, foffset, flength);
        }
        if(qlength > 0) {
            if(qualifier instanceof ByteBuffer) {
                pos = Bytes.putByteBuffer(bytes, pos, (ByteBuffer) qualifier);
            } else {
                pos = Bytes.putBytes(bytes, pos, (byte[]) qualifier, qoffset, qlength);
            }
        }
        pos = Bytes.putLong(bytes, pos, timestamp);
        pos = Bytes.putByte(bytes, pos, type.getCode());
        if(vlength > 0) {
            if(value instanceof ByteBuffer) {
                pos = Bytes.putByteBuffer(bytes, pos, (ByteBuffer) value);
            } else {
                pos = Bytes.putBytes(bytes, pos, (byte[]) value, voffset, vlength);
            }
        }
        // Add the tags after the value part
        if(tagsLength > 0) {
            pos = Bytes.putAsShort(bytes, pos, tagsLength);
            for(Tag t : tags) {
                int tlen = t.getValueLength();
                pos = Bytes.putAsShort(bytes, pos, tlen + Tag.TYPE_LENGTH_SIZE);
                pos = Bytes.putByte(bytes, pos, t.getType());
                Tag.copyValueTo(t, bytes, pos);
                pos += tlen;
            }
        }
        return bytes;
    }

    /**
     * Needed doing 'contains' on List.  Only compares the key portion, not the value.
     */
    @Override
    public boolean equals(Object other) {
        if(!(other instanceof Cell)) {
            return false;
        }
        return CellUtil.equals(this, (Cell) other);
    }

    /**
     * In line with {@link #equals(Object)}, only uses the key portion, not the value.
     */
    @Override
    public int hashCode() {
        return calculateHashForKey(this);
    }

    private int calculateHashForKey(Cell cell) {
        // pre-calculate the 3 hashes made of byte ranges
        int rowHash = Bytes.hashCode(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
        int familyHash = Bytes.hashCode(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
        int qualifierHash = Bytes.hashCode(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());

        // combine the 6 sub-hashes
        int hash = 31 * rowHash + familyHash;
        hash = 31 * hash + qualifierHash;
        hash = 31 * hash + (int) cell.getTimestamp();
        hash = 31 * hash + cell.getTypeByte();
        return hash;
    }

    //---------------------------------------------------------------------------
    //
    //  KeyValue cloning
    //
    //---------------------------------------------------------------------------

    /**
     * Clones a KeyValue.  This creates a copy, re-allocating the buffer.
     *
     * @return Fully copied clone of this KeyValue
     * @throws CloneNotSupportedException
     */
    @Override
    public KeyValue clone() throws CloneNotSupportedException {
        super.clone();
        byte[] b = new byte[this.length];
        System.arraycopy(this.bytes, this.offset, b, 0, this.length);
        KeyValue ret = new KeyValue(b, 0, b.length);
        // Important to clone the memstoreTS as well - otherwise memstore's
        // update-in-place methods (eg increment) will end up creating
        // new entries
        ret.setSequenceId(seqId);
        return ret;
    }

    /**
     * Creates a shallow copy of this KeyValue, reusing the data byte buffer.
     * http://en.wikipedia.org/wiki/Object_copy
     *
     * @return Shallow copy of this KeyValue
     */
    public KeyValue shallowCopy() {
        KeyValue shallowCopy = new KeyValue(this.bytes, this.offset, this.length);
        shallowCopy.setSequenceId(this.seqId);
        return shallowCopy;
    }

    //---------------------------------------------------------------------------
    //
    //  String representation
    //
    //---------------------------------------------------------------------------

    @Override
    public String toString() {
        if(this.bytes == null || this.bytes.length == 0) {
            return "empty";
        }
        return keyToString(this.bytes, this.offset + ROW_OFFSET, getKeyLength()) + "/vlen=" + getValueLength() + "/seqid=" + seqId;
    }

    /**
     * @param k Key portion of a KeyValue.
     * @return Key as a String, empty string if k is null.
     */
    public static String keyToString(final byte[] k) {
        if(k == null) {
            return "";
        }
        return keyToString(k, 0, k.length);
    }

    /**
     * Produces a string map for this key/value pair. Useful for programmatic use
     * and manipulation of the data stored in an WALKey, for example, printing
     * as JSON. Values are left out due to their tendency to be large. If needed,
     * they can be added manually.
     *
     * @return the Map&lt;String,?&gt; containing data from this key
     */
    public Map<String, Object> toStringMap() {
        Map<String, Object> stringMap = new HashMap<>();
        stringMap.put("row", Bytes.toStringBinary(getRowArray(), getRowOffset(), getRowLength()));
        stringMap.put("family", Bytes.toStringBinary(getFamilyArray(), getFamilyOffset(), getFamilyLength()));
        stringMap.put("qualifier", Bytes.toStringBinary(getQualifierArray(), getQualifierOffset(), getQualifierLength()));
        stringMap.put("timestamp", getTimestamp());
        stringMap.put("vlen", getValueLength());
        Iterator<Tag> tags = getTags();
        if(tags != null) {
            List<String> tagsString = new ArrayList<String>();
            while(tags.hasNext()) {
                tagsString.add(tags.next().toString());
            }
            stringMap.put("tag", tagsString);
        }
        return stringMap;
    }

    /**
     * Use for logging.
     *
     * @param b Key portion of a KeyValue.
     * @param o Offset to start of key
     * @param l Length of key.
     * @return Key as a String.
     */
    public static String keyToString(final byte[] b, final int o, final int l) {
        if(b == null)
            return "";
        int rowlength = Bytes.toShort(b, o);
        String row = Bytes.toStringBinary(b, o + Bytes.SIZEOF_SHORT, rowlength);
        int columnoffset = o + Bytes.SIZEOF_SHORT + 1 + rowlength;
        int familylength = b[columnoffset - 1];
        int columnlength = l - ((columnoffset - o) + TIMESTAMP_TYPE_SIZE);
        String family = familylength == 0 ? "" : Bytes.toStringBinary(b, columnoffset, familylength);
        String qualifier = columnlength == 0 ? "" : Bytes.toStringBinary(b, columnoffset + familylength, columnlength - familylength);
        long timestamp = Bytes.toLong(b, o + (l - TIMESTAMP_TYPE_SIZE));
        String timestampStr = humanReadableTimestamp(timestamp);
        byte type = b[o + l - 1];
        return row + "/" + family + (family != null && family.length() > 0 ? ":" : "") + qualifier + "/" + timestampStr + "/" + Type.codeToType(type);
    }

    public static String humanReadableTimestamp(final long timestamp) {
        if(timestamp == HConstants.LATEST_TIMESTAMP) {
            return "LATEST_TIMESTAMP";
        }
        if(timestamp == HConstants.OLDEST_TIMESTAMP) {
            return "OLDEST_TIMESTAMP";
        }
        return String.valueOf(timestamp);
    }

    //---------------------------------------------------------------------------
    //
    //  Public Member Accessors
    //
    //---------------------------------------------------------------------------

    /**
     * To be used only in tests where the Cells are clearly assumed to be of type KeyValue
     * and that we need access to the backing array to do some test case related assertions.
     *
     * @return The byte array backing this KeyValue.
     */
    @VisibleForTesting
    public byte[] getBuffer() {
        return this.bytes;
    }

    /**
     * @return Offset into {@link #getBuffer()} at which this KeyValue starts.
     */
    public int getOffset() {
        return this.offset;
    }

    /**
     * @return Length of bytes this KeyValue occupies in {@link #getBuffer()}.
     */
    public int getLength() {
        return length;
    }

    //---------------------------------------------------------------------------
    //
    //  Length and Offset Calculators
    //
    //---------------------------------------------------------------------------

    /**
     * Determines the total length of the KeyValue stored in the specified
     * byte array and offset.  Includes all headers.
     *
     * @param bytes  byte array
     * @param offset offset to start of the KeyValue
     * @return length of entire KeyValue, in bytes
     */
    private static int getLength(byte[] bytes, int offset) {
        int klength = ROW_OFFSET + Bytes.toInt(bytes, offset);
        int vlength = Bytes.toInt(bytes, offset + Bytes.SIZEOF_INT);
        return klength + vlength;
    }

    /**
     * @return Key offset in backing buffer..
     */
    public int getKeyOffset() {

        // TODO_MA 注释：这个key是（rowkey + family + qualifier + timestamp）的集合
        // TODO_MA 注释：ROW_OFFSET = 8
        return this.offset + ROW_OFFSET;
    }

    public String getKeyString() {
        return Bytes.toStringBinary(getBuffer(), getKeyOffset(), getKeyLength());
    }

    /**
     * @return Length of key portion.
     */
    public int getKeyLength() {

        // TODO_MA 注释：从KeyValue底层byte[]数组bytes中位置offset开始，获取一个int，也就是4B
        return Bytes.toInt(this.bytes, this.offset);
    }

    /**
     * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
     */
    @Override
    public byte[] getValueArray() {
        return bytes;
    }

    /**
     * @return the value offset
     */
    @Override
    public int getValueOffset() {

        // TODO_MA 注释：value 在 keylength, valuelength, key, 之后
        // TODO_MA 注释：valueOffset = keyOffset + keyLength
        int voffset = getKeyOffset() + getKeyLength();
        return voffset;
    }

    /**
     * @return Value length
     */
    @Override
    public int getValueLength() {

        // TODO_MA 注释：从KeyValue底层byte[]数组bytes中offset+4开始，获取一个int，也就是4B
        // TODO_MA 注释：也就是说，key length后紧跟着4B是value length
        int vlength = Bytes.toInt(this.bytes, this.offset + Bytes.SIZEOF_INT);
        return vlength;
    }

    /**
     * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
     */
    @Override
    public byte[] getRowArray() {
        return bytes;
    }

    /**
     * @return Row offset
     */
    @Override
    public int getRowOffset() {

        // TODO_MA 注释：从key的offset起，再往后略过2个字节，因为这2个字节是 rowkey length值
        return this.offset + ROW_KEY_OFFSET;
    }

    // TODO_MA 注释：根据上下这两个方法，就能读取 rowkey 的内容了。
    // TODO_MA 注释：getRowOffset() 给出读取 offset
    // TODO_MA 注释：getRowLength() 给出读取 length

    /**
     * @return Row length
     */
    @Override
    public short getRowLength() {

        // TODO_MA 注释： 从KeyOffset起，读一个short就是rowkey的长度
        // TODO_MA 注释： key的前2个字节就是 rowkey length
        return Bytes.toShort(this.bytes, getKeyOffset());
    }

    /**
     * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
     */
    @Override
    public byte[] getFamilyArray() {
        return bytes;
    }

    /**
     * @return Family offset
     */
    @Override
    public int getFamilyOffset() {

        // TODO_MA 注释： key-vlaue offset +  rowkey offset + rowkey length + rowkey length 的 length
        // TODO_MA 注释： m                +  10            + n             + 2
        // TODO_MA 注释： 整个KeyValue起始位置offset + ROW_OFFSET（Key Length + Value Length + row length） + 实际Row大小rlength + 2B（Row Length）
        return getFamilyOffset(getRowLength());
    }

    /**
     * @return Family offset
     */
    private int getFamilyOffset(int rlength) {

        // TODO_MA 注释：key-vlaue offset +  rowkey offset + rowkey length + rowkey length 的 length
        // TODO_MA 注释：ROW_KEY_OFFSET = row offset + row length
        return this.offset + ROW_KEY_OFFSET + rlength + Bytes.SIZEOF_BYTE;
    }

    /**
     * @return Family length
     */
    @Override
    public byte getFamilyLength() {

        // TODO_MA 注释：FamilyOffset - 1 即可
        return getFamilyLength(getFamilyOffset());
    }

    /**
     * @return Family length
     */
    public byte getFamilyLength(int foffset) {

        // TODO_MA 注释： 因为 FamilyLength 在 Family 的前面
        return this.bytes[foffset - 1];
    }

    /**
     * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
     */
    @Override
    public byte[] getQualifierArray() {
        return bytes;
    }

    /**
     * @return Qualifier offset
     */
    @Override
    public int getQualifierOffset() {

        // TODO_MA 注释：QualifierOffset = FamilyOffset + FamilyLength
        // TODO_MA 注释：说明 Qualifier 就在 Family 的后面
        return getQualifierOffset(getFamilyOffset());
    }

    /**
     * @return Qualifier offset
     */
    private int getQualifierOffset(int foffset) {
        return foffset + getFamilyLength(foffset);
    }

    /**
     * @return Qualifier length
     */
    @Override
    public int getQualifierLength() {
        return getQualifierLength(getRowLength(), getFamilyLength());
    }

    /**
     * @return Qualifier length
     */
    private int getQualifierLength(int rlength, int flength) {
        return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
    }

    /**
     * @return Timestamp offset
     */
    public int getTimestampOffset() {

        // TODO_MA 注释：TimestampOffset = KeyOffset + keylength - TIMESTAMP_TYPE_SIZE
        // TODO_MA 注释：TIMESTAMP_TYPE_SIZE = 9 个 字节
        return getTimestampOffset(getKeyLength());
    }

    /**
     * @param keylength Pass if you have it to save on a int creation.
     * @return Timestamp offset
     */
    private int getTimestampOffset(final int keylength) {
        return getKeyOffset() + keylength - TIMESTAMP_TYPE_SIZE;
    }

    /**
     * @return True if this KeyValue has a LATEST_TIMESTAMP timestamp.
     */
    public boolean isLatestTimestamp() {
        return Bytes.equals(getBuffer(), getTimestampOffset(), Bytes.SIZEOF_LONG, HConstants.LATEST_TIMESTAMP_BYTES, 0, Bytes.SIZEOF_LONG);
    }

    /**
     * @param now Time to set into <code>this</code> IFF timestamp ==
     *            {@link HConstants#LATEST_TIMESTAMP} (else, its a noop).
     * @return True is we modified this.
     */
    public boolean updateLatestStamp(final byte[] now) {
        if(this.isLatestTimestamp()) {
            int tsOffset = getTimestampOffset();
            System.arraycopy(now, 0, this.bytes, tsOffset, Bytes.SIZEOF_LONG);
            // clear cache or else getTimestamp() possibly returns an old value
            return true;
        }
        return false;
    }

    @Override
    public void setTimestamp(long ts) {
        Bytes.putBytes(this.bytes, this.getTimestampOffset(), Bytes.toBytes(ts), 0, Bytes.SIZEOF_LONG);
    }

    @Override
    public void setTimestamp(byte[] ts) {
        Bytes.putBytes(this.bytes, this.getTimestampOffset(), ts, 0, Bytes.SIZEOF_LONG);
    }

    //---------------------------------------------------------------------------
    //
    //  Methods that return copies of fields
    //
    //---------------------------------------------------------------------------

    /**
     * Do not use unless you have to. Used internally for compacting and testing. Use
     * {@link #getRowArray()}, {@link #getFamilyArray()}, {@link #getQualifierArray()}, and
     * {@link #getValueArray()} if accessing a KeyValue client-side.
     *
     * @return Copy of the key portion only.
     */
    public byte[] getKey() {
        int keylength = getKeyLength();
        byte[] key = new byte[keylength];
        /********
         * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
         *   注释：
         */
        System.arraycopy(getBuffer(), getKeyOffset(), key, 0, keylength);
        return key;
    }

    /**
     * @return Timestamp
     */
    @Override
    public long getTimestamp() {
        return getTimestamp(getKeyLength());
    }

    /**
     * @param keylength Pass if you have it to save on a int creation.
     * @return Timestamp
     */
    long getTimestamp(final int keylength) {
        int tsOffset = getTimestampOffset(keylength);
        return Bytes.toLong(this.bytes, tsOffset);
    }

    /**
     * @return KeyValue.TYPE byte representation
     */
    @Override
    public byte getTypeByte() {

        // TODO_MA 注释：TypeByte = this.offset + ROW_OFFSET + KeyLength - 1
        return this.bytes[this.offset + getKeyLength() - 1 + ROW_OFFSET];
    }

    /**
     * This returns the offset where the tag actually starts.
     */
    @Override
    public int getTagsOffset() {
        int tagsLen = getTagsLength();
        if(tagsLen == 0) {
            return this.offset + this.length;
        }
        return this.offset + this.length - tagsLen;
    }

    /**
     * This returns the total length of the tag bytes
     */
    @Override
    public int getTagsLength() {
        int tagsLen = this.length - (getKeyLength() + getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE);
        if(tagsLen > 0) {
            // There are some Tag bytes in the byte[].
            // So reduce 2 bytes which is added to denote the tags length
            tagsLen -= TAGS_LENGTH_SIZE;
        }
        return tagsLen;
    }

    /**
     * @return the backing array of the entire KeyValue (all KeyValue fields are in a single array)
     */
    @Override
    public byte[] getTagsArray() {
        return bytes;
    }

    /**
     * Creates a new KeyValue that only contains the key portion (the value is
     * set to be null).
     *
     * TODO only used by KeyOnlyFilter -- move there.
     *
     * @param lenAsVal replace value with the actual value length (false=empty)
     */
    public KeyValue createKeyOnly(boolean lenAsVal) {
        // KV format:  <keylen:4><valuelen:4><key:keylen><value:valuelen>
        // Rebuild as: <keylen:4><0:4><key:keylen>
        int dataLen = lenAsVal ? Bytes.SIZEOF_INT : 0;
        byte[] newBuffer = new byte[getKeyLength() + ROW_OFFSET + dataLen];
        System.arraycopy(this.bytes, this.offset, newBuffer, 0, Math.min(newBuffer.length, this.length));
        Bytes.putInt(newBuffer, Bytes.SIZEOF_INT, dataLen);
        if(lenAsVal) {
            Bytes.putInt(newBuffer, newBuffer.length - dataLen, this.getValueLength());
        }
        return new KeyValue(newBuffer);
    }

    /**
     * @param b
     * @param delimiter
     * @return Index of delimiter having started from start of <code>b</code>
     * moving rightward.
     */
    public static int getDelimiter(final byte[] b, int offset, final int length, final int delimiter) {
        if(b == null) {
            throw new IllegalArgumentException("Passed buffer is null");
        }
        int result = -1;
        for(int i = offset; i < length + offset; i++) {
            if(b[i] == delimiter) {
                result = i;
                break;
            }
        }
        return result;
    }

    /**
     * Find index of passed delimiter walking from end of buffer backwards.
     *
     * @param b
     * @param delimiter
     * @return Index of delimiter
     */
    public static int getDelimiterInReverse(final byte[] b, final int offset, final int length, final int delimiter) {
        if(b == null) {
            throw new IllegalArgumentException("Passed buffer is null");
        }
        int result = -1;
        for(int i = (offset + length) - 1; i >= offset; i--) {
            if(b[i] == delimiter) {
                result = i;
                break;
            }
        }
        return result;
    }

    /**
     * A {@link KVComparator} for <code>hbase:meta</code> catalog table
     * {@link KeyValue}s.
     *
     * @deprecated : {@link CellComparatorImpl#META_COMPARATOR} to be used. Deprecated for hbase 2.0, remove for hbase 3.0.
     */
    @Deprecated
    public static class MetaComparator extends KVComparator {
        /**
         * Compare key portion of a {@link KeyValue} for keys in <code>hbase:meta</code>
         * table.
         */
        @Override
        public int compare(final Cell left, final Cell right) {
            return PrivateCellUtil.compareKeyIgnoresMvcc(CellComparatorImpl.META_COMPARATOR, left, right);
        }

        @Override
        public int compareOnlyKeyPortion(Cell left, Cell right) {
            return compare(left, right);
        }

        @Override
        public int compareRows(byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength) {
            int leftDelimiter = getDelimiter(left, loffset, llength, HConstants.DELIMITER);
            int rightDelimiter = getDelimiter(right, roffset, rlength, HConstants.DELIMITER);
            // Compare up to the delimiter
            int lpart = (leftDelimiter < 0 ? llength : leftDelimiter - loffset);
            int rpart = (rightDelimiter < 0 ? rlength : rightDelimiter - roffset);
            int result = Bytes.compareTo(left, loffset, lpart, right, roffset, rpart);
            if(result != 0) {
                return result;
            } else {
                if(leftDelimiter < 0 && rightDelimiter >= 0) {
                    return -1;
                } else if(rightDelimiter < 0 && leftDelimiter >= 0) {
                    return 1;
                } else if(leftDelimiter < 0 && rightDelimiter < 0) {
                    return 0;
                }
            }
            // Compare middle bit of the row.
            // Move past delimiter
            leftDelimiter++;
            rightDelimiter++;
            int leftFarDelimiter = getDelimiterInReverse(left, leftDelimiter, llength - (leftDelimiter - loffset), HConstants.DELIMITER);
            int rightFarDelimiter = getDelimiterInReverse(right, rightDelimiter, rlength - (rightDelimiter - roffset), HConstants.DELIMITER);
            // Now compare middlesection of row.
            lpart = (leftFarDelimiter < 0 ? llength + loffset : leftFarDelimiter) - leftDelimiter;
            rpart = (rightFarDelimiter < 0 ? rlength + roffset : rightFarDelimiter) - rightDelimiter;
            result = super.compareRows(left, leftDelimiter, lpart, right, rightDelimiter, rpart);
            if(result != 0) {
                return result;
            } else {
                if(leftDelimiter < 0 && rightDelimiter >= 0) {
                    return -1;
                } else if(rightDelimiter < 0 && leftDelimiter >= 0) {
                    return 1;
                } else if(leftDelimiter < 0 && rightDelimiter < 0) {
                    return 0;
                }
            }
            // Compare last part of row, the rowid.
            leftFarDelimiter++;
            rightFarDelimiter++;
            result = Bytes.compareTo(left, leftFarDelimiter, llength - (leftFarDelimiter - loffset), right, rightFarDelimiter,
                    rlength - (rightFarDelimiter - roffset));
            return result;
        }

        /**
         * Don't do any fancy Block Index splitting tricks.
         */
        @Override
        public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
            return Arrays.copyOf(rightKey, rightKey.length);
        }

        /**
         * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
         * instantiate the appropriate comparator.
         * TODO: With V3 consider removing this.
         *
         * @return legacy class name for FileFileTrailer#comparatorClassName
         */
        @Override
        public String getLegacyKeyComparatorName() {
            return "org.apache.hadoop.hbase.KeyValue$MetaKeyComparator";
        }

        @Override
        protected Object clone() throws CloneNotSupportedException {
            return new MetaComparator();
        }

        /**
         * Override the row key comparison to parse and compare the meta row key parts.
         */
        @Override
        protected int compareRowKey(final Cell l, final Cell r) {
            byte[] left = l.getRowArray();
            int loffset = l.getRowOffset();
            int llength = l.getRowLength();
            byte[] right = r.getRowArray();
            int roffset = r.getRowOffset();
            int rlength = r.getRowLength();
            return compareRows(left, loffset, llength, right, roffset, rlength);
        }
    }

    /**
     * Compare KeyValues.  When we compare KeyValues, we only compare the Key
     * portion.  This means two KeyValues with same Key but different Values are
     * considered the same as far as this Comparator is concerned.
     *
     * @deprecated : Use {@link CellComparatorImpl}. Deprecated for hbase 2.0, remove for hbase 3.0.
     */
    @Deprecated
    public static class KVComparator implements RawComparator<Cell>, SamePrefixComparator<byte[]> {

        /**
         * The HFileV2 file format's trailer contains this class name.  We reinterpret this and
         * instantiate the appropriate comparator.
         * TODO: With V3 consider removing this.
         *
         * @return legacy class name for FileFileTrailer#comparatorClassName
         */
        public String getLegacyKeyComparatorName() {
            return "org.apache.hadoop.hbase.KeyValue$KeyComparator";
        }

        @Override // RawComparator
        public int compare(byte[] l, int loff, int llen, byte[] r, int roff, int rlen) {
            return compareFlatKey(l, loff, llen, r, roff, rlen);
        }


        /**
         * Compares the only the user specified portion of a Key.  This is overridden by MetaComparator.
         *
         * @param left
         * @param right
         * @return 0 if equal, &lt;0 if left smaller, &gt;0 if right smaller
         */
        protected int compareRowKey(final Cell left, final Cell right) {
            return CellComparatorImpl.COMPARATOR.compareRows(left, right);
        }

        /**
         * Compares left to right assuming that left,loffset,llength and right,roffset,rlength are
         * full KVs laid out in a flat byte[]s.
         *
         * @param left
         * @param loffset
         * @param llength
         * @param right
         * @param roffset
         * @param rlength
         * @return 0 if equal, &lt;0 if left smaller, &gt;0 if right smaller
         */
        public int compareFlatKey(byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength) {
            // Compare row
            short lrowlength = Bytes.toShort(left, loffset);
            short rrowlength = Bytes.toShort(right, roffset);
            int compare = compareRows(left, loffset + Bytes.SIZEOF_SHORT, lrowlength, right, roffset + Bytes.SIZEOF_SHORT, rrowlength);
            if(compare != 0) {
                return compare;
            }

            // Compare the rest of the two KVs without making any assumptions about
            // the common prefix. This function will not compare rows anyway, so we
            // don't need to tell it that the common prefix includes the row.
            return compareWithoutRow(0, left, loffset, llength, right, roffset, rlength, rrowlength);
        }

        public int compareFlatKey(byte[] left, byte[] right) {
            return compareFlatKey(left, 0, left.length, right, 0, right.length);
        }

        // compare a key against row/fam/qual/ts/type
        public int compareKey(Cell cell, byte[] row, int roff, int rlen, byte[] fam, int foff, int flen, byte[] col, int coff, int clen, long ts,
                byte type) {

            int compare = compareRows(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(), row, roff, rlen);
            if(compare != 0) {
                return compare;
            }
            // If the column is not specified, the "minimum" key type appears the
            // latest in the sorted order, regardless of the timestamp. This is used
            // for specifying the last key/value in a given row, because there is no
            // "lexicographically last column" (it would be infinitely long). The
            // "maximum" key type does not need this behavior.
            if(cell.getFamilyLength() + cell.getQualifierLength() == 0 && cell.getTypeByte() == Type.Minimum.getCode()) {
                // left is "bigger", i.e. it appears later in the sorted order
                return 1;
            }
            if(flen + clen == 0 && type == Type.Minimum.getCode()) {
                return -1;
            }

            compare = compareFamilies(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(), fam, foff, flen);
            if(compare != 0) {
                return compare;
            }
            compare = compareColumns(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(), col, coff, clen);
            if(compare != 0) {
                return compare;
            }
            // Next compare timestamps.
            compare = compareTimestamps(cell.getTimestamp(), ts);
            if(compare != 0) {
                return compare;
            }

            // Compare types. Let the delete types sort ahead of puts; i.e. types
            // of higher numbers sort before those of lesser numbers. Maximum (255)
            // appears ahead of everything, and minimum (0) appears after
            // everything.
            return (0xff & type) - (0xff & cell.getTypeByte());
        }

        public int compareOnlyKeyPortion(Cell left, Cell right) {
            return PrivateCellUtil.compareKeyIgnoresMvcc(CellComparatorImpl.COMPARATOR, left, right);
        }

        /**
         * Compares the Key of a cell -- with fields being more significant in this order:
         * rowkey, colfam/qual, timestamp, type, mvcc
         */
        @Override
        public int compare(final Cell left, final Cell right) {
            int compare = CellComparatorImpl.COMPARATOR.compare(left, right);
            return compare;
        }

        public int compareTimestamps(final Cell left, final Cell right) {
            return CellComparatorImpl.COMPARATOR.compareTimestamps(left, right);
        }

        /**
         * @param left
         * @param right
         * @return Result comparing rows.
         */
        public int compareRows(final Cell left, final Cell right) {
            return compareRows(left.getRowArray(), left.getRowOffset(), left.getRowLength(), right.getRowArray(), right.getRowOffset(),
                    right.getRowLength());
        }

        /**
         * Get the b[],o,l for left and right rowkey portions and compare.
         *
         * @param left
         * @param loffset
         * @param llength
         * @param right
         * @param roffset
         * @param rlength
         * @return 0 if equal, &lt;0 if left smaller, &gt;0 if right smaller
         */
        public int compareRows(byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength) {
            return Bytes.compareTo(left, loffset, llength, right, roffset, rlength);
        }

        int compareColumns(final Cell left, final short lrowlength, final Cell right, final short rrowlength) {
            return CellComparatorImpl.COMPARATOR.compareColumns(left, right);
        }

        protected int compareColumns(byte[] left, int loffset, int llength, final int lfamilylength, byte[] right, int roffset, int rlength,
                final int rfamilylength) {
            // Compare family portion first.
            int diff = Bytes.compareTo(left, loffset, lfamilylength, right, roffset, rfamilylength);
            if(diff != 0) {
                return diff;
            }
            // Compare qualifier portion
            return Bytes.compareTo(left, loffset + lfamilylength, llength - lfamilylength, right, roffset + rfamilylength, rlength - rfamilylength);
        }

        static int compareTimestamps(final long ltimestamp, final long rtimestamp) {
            // The below older timestamps sorting ahead of newer timestamps looks
            // wrong but it is intentional. This way, newer timestamps are first
            // found when we iterate over a memstore and newer versions are the
            // first we trip over when reading from a store file.
            if(ltimestamp < rtimestamp) {
                return 1;
            } else if(ltimestamp > rtimestamp) {
                return -1;
            }
            return 0;
        }

        /**
         * Overridden
         *
         * @param commonPrefix
         * @param left
         * @param loffset
         * @param llength
         * @param right
         * @param roffset
         * @param rlength
         * @return 0 if equal, &lt;0 if left smaller, &gt;0 if right smaller
         */
        @Override // SamePrefixComparator
        public int compareIgnoringPrefix(int commonPrefix, byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength) {
            // Compare row
            short lrowlength = Bytes.toShort(left, loffset);
            short rrowlength;

            int comparisonResult = 0;
            if(commonPrefix < ROW_LENGTH_SIZE) {
                // almost nothing in common
                rrowlength = Bytes.toShort(right, roffset);
                comparisonResult = compareRows(left, loffset + ROW_LENGTH_SIZE, lrowlength, right, roffset + ROW_LENGTH_SIZE, rrowlength);
            } else { // the row length is the same
                rrowlength = lrowlength;
                if(commonPrefix < ROW_LENGTH_SIZE + rrowlength) {
                    // The rows are not the same. Exclude the common prefix and compare
                    // the rest of the two rows.
                    int common = commonPrefix - ROW_LENGTH_SIZE;
                    comparisonResult = compareRows(left, loffset + common + ROW_LENGTH_SIZE, lrowlength - common, right,
                            roffset + common + ROW_LENGTH_SIZE, rrowlength - common);
                }
            }
            if(comparisonResult != 0) {
                return comparisonResult;
            }

            assert lrowlength == rrowlength;
            return compareWithoutRow(commonPrefix, left, loffset, llength, right, roffset, rlength, lrowlength);
        }

        /**
         * Compare columnFamily, qualifier, timestamp, and key type (everything
         * except the row). This method is used both in the normal comparator and
         * the "same-prefix" comparator. Note that we are assuming that row portions
         * of both KVs have already been parsed and found identical, and we don't
         * validate that assumption here.
         *
         * @param commonPrefix the length of the common prefix of the two key-values being
         *                     compared, including row length and row
         */
        private int compareWithoutRow(int commonPrefix, byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength,
                short rowlength) {
            /***
             * KeyValue Format and commonLength:
             * |_keyLen_|_valLen_|_rowLen_|_rowKey_|_famiLen_|_fami_|_Quali_|....
             * ------------------|-------commonLength--------|--------------
             */
            int commonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rowlength;

            // commonLength + TIMESTAMP_TYPE_SIZE
            int commonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + commonLength;
            // ColumnFamily + Qualifier length.
            int lcolumnlength = llength - commonLengthWithTSAndType;
            int rcolumnlength = rlength - commonLengthWithTSAndType;

            byte ltype = left[loffset + (llength - 1)];
            byte rtype = right[roffset + (rlength - 1)];

            // If the column is not specified, the "minimum" key type appears the
            // latest in the sorted order, regardless of the timestamp. This is used
            // for specifying the last key/value in a given row, because there is no
            // "lexicographically last column" (it would be infinitely long). The
            // "maximum" key type does not need this behavior.
            if(lcolumnlength == 0 && ltype == Type.Minimum.getCode()) {
                // left is "bigger", i.e. it appears later in the sorted order
                return 1;
            }
            if(rcolumnlength == 0 && rtype == Type.Minimum.getCode()) {
                return -1;
            }

            int lfamilyoffset = commonLength + loffset;
            int rfamilyoffset = commonLength + roffset;

            // Column family length.
            int lfamilylength = left[lfamilyoffset - 1];
            int rfamilylength = right[rfamilyoffset - 1];
            // If left family size is not equal to right family size, we need not
            // compare the qualifiers.
            boolean sameFamilySize = (lfamilylength == rfamilylength);
            int common = 0;
            if(commonPrefix > 0) {
                common = Math.max(0, commonPrefix - commonLength);
                if(!sameFamilySize) {
                    // Common should not be larger than Math.min(lfamilylength,
                    // rfamilylength).
                    common = Math.min(common, Math.min(lfamilylength, rfamilylength));
                } else {
                    common = Math.min(common, Math.min(lcolumnlength, rcolumnlength));
                }
            }
            if(!sameFamilySize) {
                // comparing column family is enough.
                return Bytes.compareTo(left, lfamilyoffset + common, lfamilylength - common, right, rfamilyoffset + common, rfamilylength - common);
            }
            // Compare family & qualifier together.
            final int comparison = Bytes
                    .compareTo(left, lfamilyoffset + common, lcolumnlength - common, right, rfamilyoffset + common, rcolumnlength - common);
            if(comparison != 0) {
                return comparison;
            }

            ////
            // Next compare timestamps.
            long ltimestamp = Bytes.toLong(left, loffset + (llength - TIMESTAMP_TYPE_SIZE));
            long rtimestamp = Bytes.toLong(right, roffset + (rlength - TIMESTAMP_TYPE_SIZE));
            int compare = compareTimestamps(ltimestamp, rtimestamp);
            if(compare != 0) {
                return compare;
            }

            // Compare types. Let the delete types sort ahead of puts; i.e. types
            // of higher numbers sort before those of lesser numbers. Maximum (255)
            // appears ahead of everything, and minimum (0) appears after
            // everything.
            return (0xff & rtype) - (0xff & ltype);
        }

        protected int compareFamilies(final byte[] left, final int loffset, final int lfamilylength, final byte[] right, final int roffset,
                final int rfamilylength) {
            int diff = Bytes.compareTo(left, loffset, lfamilylength, right, roffset, rfamilylength);
            return diff;
        }

        protected int compareColumns(final byte[] left, final int loffset, final int lquallength, final byte[] right, final int roffset,
                final int rquallength) {
            int diff = Bytes.compareTo(left, loffset, lquallength, right, roffset, rquallength);
            return diff;
        }

        /**
         * Compares the row and column of two keyvalues for equality
         *
         * @param left
         * @param right
         * @return True if same row and column.
         */
        public boolean matchingRowColumn(final Cell left, final Cell right) {
            short lrowlength = left.getRowLength();
            short rrowlength = right.getRowLength();

            // TsOffset = end of column data. just comparing Row+CF length of each
            if((left.getRowLength() + left.getFamilyLength() + left.getQualifierLength()) != (right.getRowLength() + right.getFamilyLength() + right
                    .getQualifierLength())) {
                return false;
            }

            if(!matchingRows(left, lrowlength, right, rrowlength)) {
                return false;
            }

            int lfoffset = left.getFamilyOffset();
            int rfoffset = right.getFamilyOffset();
            int lclength = left.getQualifierLength();
            int rclength = right.getQualifierLength();
            int lfamilylength = left.getFamilyLength();
            int rfamilylength = right.getFamilyLength();
            int diff = compareFamilies(left.getFamilyArray(), lfoffset, lfamilylength, right.getFamilyArray(), rfoffset, rfamilylength);
            if(diff != 0) {
                return false;
            } else {
                diff = compareColumns(left.getQualifierArray(), left.getQualifierOffset(), lclength, right.getQualifierArray(),
                        right.getQualifierOffset(), rclength);
                return diff == 0;
            }
        }

        /**
         * Compares the row of two keyvalues for equality
         *
         * @param left
         * @param right
         * @return True if rows match.
         */
        public boolean matchingRows(final Cell left, final Cell right) {
            short lrowlength = left.getRowLength();
            short rrowlength = right.getRowLength();
            return matchingRows(left, lrowlength, right, rrowlength);
        }

        /**
         * @param left
         * @param lrowlength
         * @param right
         * @param rrowlength
         * @return True if rows match.
         */
        private boolean matchingRows(final Cell left, final short lrowlength, final Cell right, final short rrowlength) {
            return lrowlength == rrowlength && matchingRows(left.getRowArray(), left.getRowOffset(), lrowlength, right.getRowArray(),
                    right.getRowOffset(), rrowlength);
        }

        /**
         * Compare rows. Just calls Bytes.equals, but it's good to have this encapsulated.
         *
         * @param left    Left row array.
         * @param loffset Left row offset.
         * @param llength Left row length.
         * @param right   Right row array.
         * @param roffset Right row offset.
         * @param rlength Right row length.
         * @return Whether rows are the same row.
         */
        public boolean matchingRows(final byte[] left, final int loffset, final int llength, final byte[] right, final int roffset,
                final int rlength) {
            return Bytes.equals(left, loffset, llength, right, roffset, rlength);
        }

        public byte[] calcIndexKey(byte[] lastKeyOfPreviousBlock, byte[] firstKeyInBlock) {
            byte[] fakeKey = getShortMidpointKey(lastKeyOfPreviousBlock, firstKeyInBlock);
            if(compareFlatKey(fakeKey, firstKeyInBlock) > 0) {
                LOG.error("Unexpected getShortMidpointKey result, fakeKey:" + Bytes.toStringBinary(fakeKey) + ", firstKeyInBlock:" + Bytes
                        .toStringBinary(firstKeyInBlock));
                return firstKeyInBlock;
            }
            if(lastKeyOfPreviousBlock != null && compareFlatKey(lastKeyOfPreviousBlock, fakeKey) >= 0) {
                LOG.error("Unexpected getShortMidpointKey result, lastKeyOfPreviousBlock:" + Bytes
                        .toStringBinary(lastKeyOfPreviousBlock) + ", fakeKey:" + Bytes.toStringBinary(fakeKey));
                return firstKeyInBlock;
            }
            return fakeKey;
        }

        /**
         * This is a HFile block index key optimization.
         *
         * @param leftKey
         * @param rightKey
         * @return 0 if equal, &lt;0 if left smaller, &gt;0 if right smaller
         * @deprecated Since 0.99.2;
         */
        @Deprecated
        public byte[] getShortMidpointKey(final byte[] leftKey, final byte[] rightKey) {
            if(rightKey == null) {
                throw new IllegalArgumentException("rightKey can not be null");
            }
            if(leftKey == null) {
                return Arrays.copyOf(rightKey, rightKey.length);
            }
            if(compareFlatKey(leftKey, rightKey) >= 0) {
                throw new IllegalArgumentException("Unexpected input, leftKey:" + Bytes.toString(leftKey) + ", rightKey:" + Bytes.toString(rightKey));
            }

            short leftRowLength = Bytes.toShort(leftKey, 0);
            short rightRowLength = Bytes.toShort(rightKey, 0);
            int leftCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + leftRowLength;
            int rightCommonLength = ROW_LENGTH_SIZE + FAMILY_LENGTH_SIZE + rightRowLength;
            int leftCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + leftCommonLength;
            int rightCommonLengthWithTSAndType = TIMESTAMP_TYPE_SIZE + rightCommonLength;
            int leftColumnLength = leftKey.length - leftCommonLengthWithTSAndType;
            int rightColumnLength = rightKey.length - rightCommonLengthWithTSAndType;
            // rows are equal
            if(leftRowLength == rightRowLength && compareRows(leftKey, ROW_LENGTH_SIZE, leftRowLength, rightKey, ROW_LENGTH_SIZE,
                    rightRowLength) == 0) {
                // Compare family & qualifier together.
                int comparison = Bytes.compareTo(leftKey, leftCommonLength, leftColumnLength, rightKey, rightCommonLength, rightColumnLength);
                // same with "row + family + qualifier", return rightKey directly
                if(comparison == 0) {
                    return Arrays.copyOf(rightKey, rightKey.length);
                }
                // "family + qualifier" are different, generate a faked key per rightKey
                byte[] newKey = Arrays.copyOf(rightKey, rightKey.length);
                Bytes.putLong(newKey, rightKey.length - TIMESTAMP_TYPE_SIZE, HConstants.LATEST_TIMESTAMP);
                Bytes.putByte(newKey, rightKey.length - TYPE_SIZE, Type.Maximum.getCode());
                return newKey;
            }
            // rows are different
            short minLength = leftRowLength < rightRowLength ? leftRowLength : rightRowLength;
            short diffIdx = 0;
            while(diffIdx < minLength && leftKey[ROW_LENGTH_SIZE + diffIdx] == rightKey[ROW_LENGTH_SIZE + diffIdx]) {
                diffIdx++;
            }
            byte[] newRowKey = null;
            if(diffIdx >= minLength) {
                // leftKey's row is prefix of rightKey's.
                newRowKey = new byte[diffIdx + 1];
                System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
            } else {
                int diffByte = leftKey[ROW_LENGTH_SIZE + diffIdx];
                if((0xff & diffByte) < 0xff && (diffByte + 1) < (rightKey[ROW_LENGTH_SIZE + diffIdx] & 0xff)) {
                    newRowKey = new byte[diffIdx + 1];
                    System.arraycopy(leftKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx);
                    newRowKey[diffIdx] = (byte) (diffByte + 1);
                } else {
                    newRowKey = new byte[diffIdx + 1];
                    System.arraycopy(rightKey, ROW_LENGTH_SIZE, newRowKey, 0, diffIdx + 1);
                }
            }
            return new KeyValue(newRowKey, null, null, HConstants.LATEST_TIMESTAMP, Type.Maximum).getKey();
        }

        @Override
        protected Object clone() throws CloneNotSupportedException {
            super.clone();
            return new KVComparator();
        }

    }

    /**
     * @param in Where to read bytes from.  Creates a byte array to hold the KeyValue
     *           backing bytes copied from the steam.
     * @return KeyValue created by deserializing from <code>in</code> OR if we find a length
     * of zero, we will return null which can be useful marking a stream as done.
     * @throws IOException
     */
    public static KeyValue create(final DataInput in) throws IOException {
        return create(in.readInt(), in);
    }

    /**
     * Create a KeyValue reading <code>length</code> from <code>in</code>
     *
     * @param length
     * @param in
     * @return Created KeyValue OR if we find a length of zero, we will return null which
     * can be useful marking a stream as done.
     * @throws IOException
     */
    public static KeyValue create(int length, final DataInput in) throws IOException {

        if(length <= 0) {
            if(length == 0)
                return null;
            throw new IOException("Failed read " + length + " bytes, stream corrupt?");
        }

        // This is how the old Writables.readFrom used to deserialize.  Didn't even vint.
        byte[] bytes = new byte[length];
        in.readFully(bytes);
        return new KeyValue(bytes, 0, length);
    }

    /**
     * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable.
     *
     * @param kv
     * @param out
     * @return Length written on stream
     * @throws IOException
     * @see #create(DataInput) for the inverse function
     */
    public static long write(final KeyValue kv, final DataOutput out) throws IOException {
        // This is how the old Writables write used to serialize KVs.  Need to figure way to make it
        // work for all implementations.
        int length = kv.getLength();
        out.writeInt(length);
        out.write(kv.getBuffer(), kv.getOffset(), length);
        return (long) length + Bytes.SIZEOF_INT;
    }

    /**
     * Write out a KeyValue in the manner in which we used to when KeyValue was a Writable but do
     * not require a {@link DataOutput}, just take plain {@link OutputStream}
     * Named <code>oswrite</code> so does not clash with {@link #write(KeyValue, DataOutput)}
     *
     * @param kv
     * @param out
     * @param withTags
     * @return Length written on stream
     * @throws IOException
     * @see #create(DataInput) for the inverse function
     * @see #write(KeyValue, DataOutput)
     * @see KeyValueUtil#oswrite(Cell, OutputStream, boolean)
     * @deprecated As of release 2.0.0, this will be removed in HBase 3.0.0.
     * Instead use {@link #write(OutputStream, boolean)}
     */
    @Deprecated
    public static long oswrite(final KeyValue kv, final OutputStream out, final boolean withTags) throws IOException {
        ByteBufferUtils.putInt(out, kv.getSerializedSize(withTags));
        return (long) kv.write(out, withTags) + Bytes.SIZEOF_INT;
    }

    @Override
    public int write(OutputStream out, boolean withTags) throws IOException {
        int len = getSerializedSize(withTags);
        out.write(this.bytes, this.offset, len);
        return len;
    }

    @Override
    public int getSerializedSize(boolean withTags) {
        if(withTags) {
            return this.length;
        }
        return this.getKeyLength() + this.getValueLength() + KEYVALUE_INFRASTRUCTURE_SIZE;
    }

    @Override
    public int getSerializedSize() {
        return this.length;
    }

    @Override
    public void write(ByteBuffer buf, int offset) {
        ByteBufferUtils.copyFromArrayToBuffer(buf, offset, this.bytes, this.offset, this.length);
    }

    /**
     * Avoids redundant comparisons for better performance.
     *
     * TODO get rid of this wart
     */
    public interface SamePrefixComparator<T> {
        /**
         * Compare two keys assuming that the first n bytes are the same.
         *
         * @param commonPrefix How many bytes are the same.
         */
        int compareIgnoringPrefix(int commonPrefix, byte[] left, int loffset, int llength, byte[] right, int roffset, int rlength);
    }

    /**
     * HeapSize implementation
     *
     * We do not count the bytes in the rowCache because it should be empty for a KeyValue in the
     * MemStore.
     */
    @Override
    public long heapSize() {
        /*
         * Deep object overhead for this KV consists of two parts. The first part is the KV object
         * itself, while the second part is the backing byte[]. We will only count the array overhead
         * from the byte[] only if this is the first KV in there.
         */
        return ClassSize.align(FIXED_OVERHEAD) + (offset == 0 ? ClassSize.sizeOfByteArray(length)  // count both length and object overhead
                : length);                           // only count the number of bytes
    }

    /**
     * A simple form of KeyValue that creates a keyvalue with only the key part of the byte[]
     * Mainly used in places where we need to compare two cells.  Avoids copying of bytes
     * In places like block index keys, we need to compare the key byte[] with a cell.
     * Hence create a Keyvalue(aka Cell) that would help in comparing as two cells
     */
    public static class KeyOnlyKeyValue extends KeyValue {
        private short rowLen = -1;

        public KeyOnlyKeyValue() {

        }

        public KeyOnlyKeyValue(byte[] b) {
            this(b, 0, b.length);
        }

        public KeyOnlyKeyValue(byte[] b, int offset, int length) {
            this.bytes = b;
            this.length = length;
            this.offset = offset;
            this.rowLen = Bytes.toShort(this.bytes, this.offset);
        }

        public void set(KeyOnlyKeyValue keyOnlyKeyValue) {
            this.bytes = keyOnlyKeyValue.bytes;
            this.length = keyOnlyKeyValue.length;
            this.offset = keyOnlyKeyValue.offset;
            this.rowLen = keyOnlyKeyValue.rowLen;
        }

        public void clear() {
            rowLen = -1;
            bytes = null;
            offset = 0;
            length = 0;
        }

        @Override
        public int getKeyOffset() {
            return this.offset;
        }

        /**
         * A setter that helps to avoid object creation every time and whenever
         * there is a need to create new KeyOnlyKeyValue.
         *
         * @param key
         * @param offset
         * @param length
         */
        public void setKey(byte[] key, int offset, int length) {
            this.bytes = key;
            this.offset = offset;
            this.length = length;
            this.rowLen = Bytes.toShort(this.bytes, this.offset);
        }

        @Override
        public byte[] getKey() {
            int keylength = getKeyLength();
            byte[] key = new byte[keylength];
            System.arraycopy(this.bytes, getKeyOffset(), key, 0, keylength);
            return key;
        }

        @Override
        public byte[] getRowArray() {
            return bytes;
        }

        @Override
        public int getRowOffset() {
            return getKeyOffset() + Bytes.SIZEOF_SHORT;
        }

        @Override
        public byte[] getFamilyArray() {
            return bytes;
        }

        @Override
        public byte getFamilyLength() {
            return this.bytes[getFamilyOffset() - 1];
        }

        @Override
        public int getFamilyOffset() {
            return this.offset + Bytes.SIZEOF_SHORT + getRowLength() + Bytes.SIZEOF_BYTE;
        }

        @Override
        public byte[] getQualifierArray() {
            return bytes;
        }

        @Override
        public int getQualifierLength() {
            return getQualifierLength(getRowLength(), getFamilyLength());
        }

        @Override
        public int getQualifierOffset() {
            return getFamilyOffset() + getFamilyLength();
        }

        @Override
        public int getKeyLength() {
            return length;
        }

        @Override
        public short getRowLength() {
            return rowLen;
        }

        @Override
        public byte getTypeByte() {
            return this.bytes[this.offset + getKeyLength() - 1];
        }

        private int getQualifierLength(int rlength, int flength) {
            return getKeyLength() - (int) getKeyDataStructureSize(rlength, flength, 0);
        }

        @Override
        public long getTimestamp() {
            int tsOffset = getTimestampOffset();
            return Bytes.toLong(this.bytes, tsOffset);
        }

        @Override
        public int getTimestampOffset() {
            return getKeyOffset() + getKeyLength() - TIMESTAMP_TYPE_SIZE;
        }

        @Override
        public byte[] getTagsArray() {
            return HConstants.EMPTY_BYTE_ARRAY;
        }

        @Override
        public int getTagsOffset() {
            return 0;
        }

        @Override
        public byte[] getValueArray() {
            throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
        }

        @Override
        public int getValueOffset() {
            throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
        }

        @Override
        public int getValueLength() {
            throw new IllegalArgumentException("KeyOnlyKeyValue does not work with values.");
        }

        @Override
        public int getTagsLength() {
            return 0;
        }

        @Override
        public String toString() {
            if(this.bytes == null || this.bytes.length == 0) {
                return "empty";
            }
            return keyToString(this.bytes, this.offset, getKeyLength()) + "/vlen=0/mvcc=0";
        }

        @Override
        public int hashCode() {
            return super.hashCode();
        }

        @Override
        public boolean equals(Object other) {
            return super.equals(other);
        }

        @Override
        public long heapSize() {
            return super.heapSize() + Bytes.SIZEOF_SHORT;
        }

        @Override
        public int write(OutputStream out, boolean withTags) throws IOException {
            // This type of Cell is used only to maintain some internal states. We never allow this type
            // of Cell to be returned back over the RPC
            throw new IllegalStateException("A reader should never return this type of a Cell");
        }
    }

    @Override
    public ExtendedCell deepClone() {
        byte[] copy = Bytes.copy(this.bytes, this.offset, this.length);
        KeyValue kv = new KeyValue(copy, 0, copy.length);
        kv.setSequenceId(this.getSequenceId());
        return kv;
    }
}
